package com.aotain.coeus;

import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.HashMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;

import com.aotain.common.CommonFunction;
import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.IDCIPSegment;
import com.aotain.apollo.IPDatabase;
import com.aotain.apollo.IPSearch;

/**
 * 域名IP 关系 结合备案信息分析
 * Created by turk 2016-03-03
 * @author Administrator
 *
 */
public class DomainIPAly {

	public static void main(String[] args) {
		   //System.out.println("0###OK################################");
		
		if (args.length != 4){
		   System.err.printf("Usage: <SourcePath> <OutputPath> <DBHOST> <ProviceID>");
		   System.exit(1);
		}          
		
		//HTTPDCSpark spark = new HTTPDCSpark();
		int nexit = SparkFunc(args);
		   //System.out.println("1###OK################################");
		System.exit(nexit);
	}
	
	private static int SparkFunc(String[] args)
	{
		/**
		 * 表结构
		 * domainname  域名
		 * ip IP
		 * num
		 */
		try
		 {
			String dbhost = args[2];
			String DBJSON="{\"driverClassName\":\"com.mysql.jdbc.Driver\","
					+ "\"maxActive\":\"200\",\"maxIdle\":\"50\",\"maxWait\":\"10000\","
					+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
					+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:mysql://"+dbhost+":3306/SDS\","
					+ "\"userName\":\"sds\",\"validationQuery\":\"SELECT SYSDATE()\"}";
					
			//读取域名备案数据
			ApolloConfig ap = new ApolloConfig(DBJSON);
			final HashMap<String,String> domainMap = ap.getDomainRecord();
			 
			//读取机房IP段配置信息
			final IDCIPSegment ipseg = new IDCIPSegment();
			ipseg.GetIDCIPSegment(DBJSON);
			 
//			Logger.getRootLogger().info("######----INIT IP DB---#####");
//			final HashMap<String,String> IPRecordMap = ap.getIPRecord();
//			Logger.getRootLogger().info("######----END IP DB---#####");
			 
			 SparkConf sparkConf = new SparkConf().setAppName("DomainIPAly");
			 Configuration config = new Configuration();
			 JavaSparkContext ctx = new JavaSparkContext(sparkConf);
			 
			 String sourcePath = args[0];
			 String outputPath = args[1];
			 
			 final int ProviceID = Integer.parseInt(args[3]);
			 
			 System.out.println("sourcePath: " + sourcePath);
			 
		
			//目标目录维护
			//String targetPath = args[1];
			System.out.println("targetPath: " + outputPath);
			FileSystem fsTarget = FileSystem.get(URI.create(outputPath),config);
			Path pathTarget = new Path(outputPath);
			if(fsTarget.exists(pathTarget))
			{
				fsTarget.delete(pathTarget, true);
				System.out.println("Delete path " + outputPath);
			}
			final IPSearch ipsearch = new IPSearch(DBJSON);
		
		 
		 
		 JavaPairRDD<LongWritable,Text> tt = ctx.newAPIHadoopFile(sourcePath,TextInputFormat.class,
				 LongWritable.class, Text.class,config);
		 JavaPairRDD<String, String> s = tt.mapToPair(new PairFunction<Tuple2<LongWritable,Text>, 
				 String,String>() {
		    	
			/**
			 * serialVersionUID
			 */
			 private static final long serialVersionUID = 605833049875773474L;

			@Override
			public Tuple2<String, String> call(Tuple2<LongWritable, Text> arg0)
					throws Exception {
				// TODO Auto-generated method stub
				//从hive表中获取一行记录 , 分割字段
				
				//昨天
				Calendar c = Calendar.getInstance();
				c.add(Calendar.DATE, -1);//				
				SimpleDateFormat sdf  = new SimpleDateFormat("yyyy-MM-dd");
				String strDateD = sdf.format(c.getTime());
				
				String[] items = arg0._2.toString().split(",",-1);
				if(items.length < 3)
				{
					Logger.getLogger(DomainIPAly.class).error("items size < 3:" + arg0._2.toString());
					return null;
				}
				String idc = items[0];
				String domainname = items[1].trim();
				String ip = items[2].trim();
				
				if(CommonFunction.isRegex(ip, "^[0-9]*[1-9][0-9]*$"))
				{//判断是否为数字
					ip = CommonFunction.int2ip(Long.parseLong(ip));
				}
				
				String strKey = "";
				
				//IDCNAME#COMPANY
				String idcAndCompany = ipseg.getIDCNameByIP(ip);
				
//				if(idcAndCompany==null) {
//					//Logger.getLogger(DomainIPAly.class).error("NONE IDC IP " + idc);
//					return null;
//				}
				
				String idcname = idcAndCompany.split("#",-1)[0];
				String ipsegCompany = idcAndCompany.split("#",-1)[1];
				
				IPDatabase ipinfo = ipsearch.getStartIP(ip);
				
				if(ipinfo == null || ipinfo.getProviceID() != ProviceID)
				{
					//Logger.getLogger(DomainIPAly.class).info("##IP:"+ip);
					return null;
				}
				
				//全域名
				
				String rexp = "([1-9]|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])(\\.(\\d|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])){3}";
				
				String[] domainValues = domainname.split(":");
				if(domainValues.length > 1)
				{
					domainname = domainValues[0];
				}
				
				String[] strArr = domainname.split("\\.",-1);
				
				String IP = CommonFunction.findByRegex(domainname, rexp, 0);
				
				String row = "";
				//WEBRECORDNUM|WEBSITENAME|IP|COMPANY|COMPANYTYPE|CITYNAME|RECORDNUM
				
				String domainRecord = "";
				String ips = "";
				String webrecordnum = "";
				String websitename = "";
				String company = "";
				String companytype = "";
				String cityname = "";
				String recordnum = "";
				
				
				if(strArr.length >= 2)
				{//根域名//顶级域名
					 //判断是否为IP地址
					 if(IP != null)
					 {//如果是IP，返回
						 //return null;
						 strKey = IP;
					 }
					 
					 else if(CommonFunction.IsSpecialRoot(domainname) && strArr.length >= 3)
					 {//特殊处理例如 .com.cn  .com.hk
						 strKey = String.format("%s.%s.%s", strArr[strArr.length-3],
								 strArr[strArr.length-2],strArr[strArr.length-1]);
					 }
					 else if(CommonFunction.IsSpecialRoot2Level(domainname))
					 {
						 strKey = String.format("%s.%s", 
								 strArr[strArr.length-2],strArr[strArr.length-1]);
					 }
					 else
					 {
						 return null;
					 }
					 
					 boolean isRecord = false;
					 
					 if(domainMap.containsKey(strKey))
					 {	//发现域名备案信息
						//WEBRECORDNUM|WEBSITENAME|IP|COMPANY|COMPANYTYPE|CITYNAME|RECORDNUM
						 domainRecord = domainMap.get(strKey);
						 String[] cols = domainRecord.split("\\|");
						 if(cols.length == 7)
						 {
							 webrecordnum = cols[0];
							 websitename = cols[1];
							 ips = cols[2];
							 company = cols[3];
							 companytype = cols[4];
							 cityname = cols[5];
							 recordnum = cols[6];
							 
							 String[] arrIPs = ips.split(";");
							 
							 long currentIP = CommonFunction.ip2int(ip);
							 for(String sIP : arrIPs)
							 {//218.22.248.16-218.22.248.23;218.22.248.0-218.22.248.15
								 //判断IP 是否在备案信息中
								 if(sIP.contains("-"))
								 {
									 long startIP =  CommonFunction.ip2int(sIP.split("-",-1)[0]);
									 long endIP =  CommonFunction.ip2int(sIP.split("-",-1)[1]);
									
									 if(currentIP >= startIP && currentIP <= endIP)
									 {
										 isRecord = true;
									 }
								 }
								 else
								 {
									 long startIP =  CommonFunction.ip2int(sIP);
									 if(startIP == currentIP)
										 isRecord = true;
								 }
							 }
						 }
					 }
					 
					 /*	V-S2 20160511 add by turk
						 *  IDCNAME	
							ACCESSAREA->CITYNAME
							IPAREA	
							RECORDNUM	
							IPCOMPANY	
							HTTPNUM*/	
					 
					 //0：IP未备案   1：IP已备案
					 String IPRecord = "0";
					 if(isRecord)
						 IPRecord = "1";
					 
					 String iparea = ipinfo.getCityName();
					 String ipcompany = "";//IPRecordMap.get(ip);
					 //REPORTDATE|IDC|DOMAIN|IP|WEBRECORDNUM|
					 //WEBSITENAME|COMPANY|COMPANYTYPE|ISIPRECORD|IDCNAME|
					 //CITYNAME|IPAREA|RECORDNUM|IPCOMPANY|IPSEGCOMPANY
					 row = String.format("%s|%s|%s|%s|%s|"
							 		+ "%s|%s|%s|%s|%s|"
							 		+ "%s|%s|%s|%s|%s|", 
							 strDateD,idc, strKey, ip, webrecordnum,
							 websitename, company, companytype, IPRecord,idcname, 
							 cityname, iparea, recordnum,ipcompany,ipsegCompany);
					 return new Tuple2<String, String>(row,"");
				 }
				
				return null;
				 
//				 if(strArr.length >= 3 && !domain.contains("www.") && IP == null)
//				 {
//					 //二级
//					 //排除www
//					 //判断是否为IP地址
//					 
//					 if(IsSpecialRoot(domain) && strArr.length == 3)
//						 return;//对于com.cn的域名特殊判断
//					 
//					 if(IsSpecialRoot(domain) && strArr.length > 3)
//					 {
//						 strKey = String.format("%s.%s.%s.%s_2", 
//								 strArr[strArr.length-4],
//								 strArr[strArr.length-3],
//								 strArr[strArr.length-2],strArr[strArr.length-1]);
//					 }
//					 else
//					 {
//						 strKey = String.format("%s.%s.%s_2", strArr[strArr.length-3],
//							 strArr[strArr.length-2],strArr[strArr.length-1]);
//					 }
//					 context.write(new Text(strKey), new IntWritable(1));
//				 }
//				
				
				
				
				
			}
		 });
		 
		 JavaPairRDD<String, String> result = s.filter(new Function<Tuple2<String, String>, Boolean>(){

			@Override
			public Boolean call(Tuple2<String, String> arg0) throws Exception {
				// TODO Auto-generated method stub
				if(arg0 == null)
					return false;
				else
					return true;
			}}).distinct();
		 
		result.saveAsHadoopFile(outputPath, Text.class, Text.class, TextOutputFormat.class);
		    	
		ctx.stop();
		return 0;
		 }
		 catch(Exception ex)
		 {
			 ex.printStackTrace();
			 return 1;
		 }
		
	}
	


	
}
